package com.talent.service.iot.mqtt.handler;

import com.talent.service.iot.mqtt.config.Constant;
import com.talent.service.iot.mqtt.model.request.TlMqttDisconnectReq;
import com.talent.service.iot.mqtt.model.request.TlMqttPublishReq;
import com.talent.service.iot.mqtt.model.variable.TlMqttPublishVariableHead;
import com.talent.service.iot.mqtt.service.IMqttSessionService;
import com.talent.service.iot.mqtt.service.IMqttSubscriptionService;
import io.netty.channel.Channel;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.SimpleChannelInboundHandler;
import io.netty.util.AttributeKey;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

/**
 * @ClassName：ConnectHandler
 * @Author: hszhou
 * @Date: 2024/11/25 11:06
 * @Description: 必须描述类做什么事情, 实现什么功能
 */
@ChannelHandler.Sharable
@Component
public class TlMqttDisconnectHandler extends SimpleChannelInboundHandler<TlMqttDisconnectReq> {

    private static final  Logger logger = LoggerFactory.getLogger(TlMqttDisconnectHandler.class);

    private final IMqttSubscriptionService subscriptionService;

    private final IMqttSessionService sessionService;

    public TlMqttDisconnectHandler(IMqttSubscriptionService subscriptionService,
                                   IMqttSessionService sessionService){
        this.subscriptionService=subscriptionService;
        this.sessionService = sessionService;
    }
    @Override
    protected void channelRead0(ChannelHandlerContext ctx, TlMqttDisconnectReq msg) throws Exception {
        //断开标志位设置为true 这样就不发生遗嘱消息了
        ctx.channel().attr(AttributeKey.valueOf(Constant.DISCONNECT)).set(true);
        ctx.channel().close();
        logger.info("获取到断开连接指令");
    }

    @Override
    public void channelInactive(ChannelHandlerContext ctx) throws Exception {

        Boolean disconnectFlag= (Boolean)ctx.channel().attr(AttributeKey.valueOf(Constant.DISCONNECT)).get();

        Object clientIdObj = ctx.channel().attr(AttributeKey.valueOf(Constant.CLIENT_ID)).get();
        if(null==clientIdObj){

            ctx.channel().close();
            return;
        }
        String clientId = clientIdObj.toString();
        logger.info("client 【{}】 disconnect",clientId);

        //将客户端的会话通道置空
        sessionService.find(clientId)
                .doOnSuccess(session -> ctx.channel().close())
                .subscribe(session -> {
                    session.setChannel(null);
                    //起一个定时任务  n分钟后如何还没有连接 就直接断开
                   if(disconnectFlag){
                       return;
                   }
                   logger.debug("client 【{}】 disconnect unsafe send willMessage if has",clientId);
                   TlMqttPublishReq willMessage = session.getWillMessage();
                   TlMqttPublishVariableHead variableHead = willMessage.getVariableHead();
                   String topic = variableHead.getTopic();
                   subscriptionService.find(topic)
                            .subscribe(sub-> sessionService.find(clientId)
                                    .subscribe(subSession -> {
                                        Channel channel = subSession.getChannel();
                                        if (channel != null) {
                                            channel.writeAndFlush(willMessage);
                                        }
                                    }));

        });

    }
}
